/*
 * Copyright (c) 2022. China Mobile (SuZhou) Software Technology Co.,Ltd. All rights reserved.
 * Lakehouse is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */

package com.chinamobile.cmss.lakehouse.service.engine.impl;

import com.chinamobile.cmss.lakehouse.common.dto.engine.LakehouseResponse;
import com.chinamobile.cmss.lakehouse.common.enums.HttpStatus;
import com.chinamobile.cmss.lakehouse.service.engine.ILogService;

import java.io.BufferedReader;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FilenameUtils;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class SparkLogServiceImpl implements ILogService {

    @Override
    public LakehouseResponse checkStartSuccess(String path) {
        Path logFile = Paths.get(FilenameUtils.normalize(path));
        if (!Files.exists(logFile) || Files.isDirectory(logFile)) {
            return new LakehouseResponse(HttpStatus.BAD_REQUEST, "File not exist!");
        }
        try (BufferedReader reader = Files.newBufferedReader(logFile, StandardCharsets.UTF_8)) {
            String line;
            int lineIndex = 1;
            while ((line = reader.readLine()) != null && lineIndex < 150) {
                if (line.contains("(phase: Pending)")) {
                    return LakehouseResponse.success();
                }
            }
            return new LakehouseResponse(HttpStatus.INTERNAL_SERVER_ERROR, "Spark task failed!");
        } catch (IOException ioException) {
            log.error("{}", ioException.getMessage(), ioException);
            return new LakehouseResponse(HttpStatus.INTERNAL_SERVER_ERROR, ioException.getCause());
        }
    }
}
